import pandas as pd
import json
import os
import random
from datetime import datetime
import re


def parse_kafka_value_content(value_str):
    """
    解析Kafka消息value内容，提取Header中的字段和HTTP响应信息
    """
    result = {}

    if not isinstance(value_str, str) or not value_str.strip():
        return result

    try:
        # 提取Header部分
        header_match = re.search(r'Header:(\{[^}]+\})', value_str)
        if header_match:
            header_json = header_match.group(1)
            try:
                header_data = json.loads(header_json)
                # 提取Header中的字段
                for key, value in header_data.items():
                    result[key] = value
            except json.JSONDecodeError as e:
                print(f"JSON解析失败: {e}")

        # 提取完整的HTTP请求行 (如 GET /4e0e3b2f9a33cf694a5a.ttf HTTP/1.1)
        http_request_match = re.search(r'(GET|POST|PUT|DELETE|HEAD|OPTIONS|PATCH)\s+([^\s]+)\s+(HTTP/[\d.]+)',
                                       value_str)
        if http_request_match:
            method = http_request_match.group(1)
            uri = http_request_match.group(2)
            version = http_request_match.group(3)
            result['http_request_line'] = f"{method} {uri} {version}"

        # 提取HTTP响应码和状态
        http_status_match = re.search(r'HTTP/1\.1 (\d+) ([^\r\n]+)', value_str)
        if http_status_match:
            result['_http_status_code'] = http_status_match.group(1)
            result['_http_status_text'] = http_status_match.group(2)

    except Exception as e:
        print(f"解析过程中出错: {e}")
        result['parse_error'] = str(e)

    return result


def process_kafka_excel_file(input_file, required_columns):
    """
    处理包含Kafka消息的Excel文件（无表头格式）
    """
    print(f"开始处理文件: {input_file}")

    if not os.path.exists(input_file):
        print(f"错误: 文件不存在: {input_file}")
        return None

    try:
        # 读取Excel文件（无表头）
        print("正在读取Excel文件...")
        df = pd.read_excel(input_file, header=None)
        print(f"成功读取文件: {input_file}")
        print(f"数据行数: {len(df)}")
        print(f"列数: {len(df.columns)}")

        # 显示前几行数据结构
        if len(df) > 0:
            print("前2行数据示例:")
            for i in range(min(2, len(df))):
                row_data = df.iloc[i]
                print(
                    f"  第{i + 1}行: A列={row_data[0] if len(row_data) > 0 else 'N/A'}, B列={row_data[1] if len(row_data) > 1 else 'N/A'}, C列={str(row_data[2])[:50] if len(row_data) > 2 else 'N/A'}...")

        # 检查列数是否足够
        if len(df.columns) < 3:
            print("错误: 文件列数不足，至少需要3列(A,B,C)")
            return None

        # 解析每行数据
        parsed_results = []
        print("开始解析数据...")

        for index, row in df.iterrows():
            # 提取各列数据
            offset_data = row[0] if len(row) > 0 else ""  # A列 - 偏移量
            key_data = row[1] if len(row) > 1 else ""  # B列 - Key
            value_data = row[2] if len(row) > 2 else ""  # C列 - Value

            # 处理空值
            if pd.isna(offset_data):
                offset_data = ""
            if pd.isna(key_data):
                key_data = ""
            if pd.isna(value_data):
                value_data = ""

            # 解析value内容
            parsed_data = parse_kafka_value_content(str(value_data))

            # 显示前几行解析示例
            if index < 2:
                print(f"第{index + 1}行解析示例:")
                print(f"  offset: {offset_data}")
                print(f"  key: {key_data}")
                print(f"  解析字段: {list(parsed_data.items())[:3]}")

            # 构建结果数据
            extracted_data = {}
            for col in required_columns:
                if col == 'offset':
                    extracted_data[col] = offset_data
                elif col == 'key':
                    extracted_data[col] = key_data
                else:
                    extracted_data[col] = parsed_data.get(col, '')

            parsed_results.append(extracted_data)

        # 创建结果DataFrame
        result_df = pd.DataFrame(parsed_results)
        print(f"解析完成，共处理 {len(parsed_results)} 行数据")
        return result_df

    except Exception as e:
        print(f"处理文件时出错: {e}")
        import traceback
        traceback.print_exc()
        return None


def save_parsed_results(df, output_dir, required_columns):
    """
    保存解析结果到Excel文件
    """
    if df is None or df.empty:
        print("没有数据需要保存")
        return False

    try:
        # 确保输出目录存在
        os.makedirs(output_dir, exist_ok=True)

        # 生成文件名 kaf_日期_4位随机数.xlsx
        current_date = datetime.now().strftime('%m%d')
        random_number = random.randint(1000, 9999)
        filename = f"kaf_{current_date}_{random_number}.xlsx"
        output_path = os.path.join(output_dir, filename)

        # 调整四元组排序为 sip spt dip dpt
        final_columns = []
        
        # 按指定顺序添加四元组列
        tuple_columns = ['sip', 'spt', 'dip', 'dpt']
        for col in tuple_columns:
            if col in df.columns:
                final_columns.append(col)
        
        # 添加其他必需列
        for col in required_columns:
            if col not in final_columns:
                final_columns.append(col)
        
        # 添加剩余列
        for col in df.columns:
            if col not in final_columns:
                final_columns.append(col)

        # 重新排列DataFrame列
        df = df[final_columns]

        # 保存到Excel
        df.to_excel(output_path, index=False)
        print(f"结果已保存到: {output_path}")
        print(f"保存行数: {len(df)}")
        print(f"列名: {list(df.columns)}")
        return True

    except Exception as e:
        print(f"保存文件时出错: {e}")
        import traceback
        traceback.print_exc()
        return False


def main():
    # 配置
    input_file = r"D:\DPI测试\转储测试\Data\1.xlsx"
    output_dir = r"D:\DPI测试\转储测试\Data"

    # 定义需要提取的列（调整四元组排序为 sip spt dip dpt）
    required_columns = [
        'offset',  # A列 - 偏移量
        'key',     # B列 - Key
        'sip', 'spt', 'dip', 'dpt',  # 四元组按指定顺序
        'cid', 'dir', 'proto',
        'http_request_line'  # HTTP请求行
    ]

    print("Kafka消息解析工具（无表头格式）")
    print("=" * 40)
    print(f"输入文件: {input_file}")
    print(f"输出目录: {output_dir}")
    print(f"提取列: {required_columns}")

    if not os.path.exists(input_file):
        print(f"错误: 输入文件不存在: {input_file}")
        return

    # 处理Excel文件
    result_df = process_kafka_excel_file(input_file, required_columns)

    if result_df is not None and not result_df.empty:
        # 显示结果统计
        print("\n解析结果统计:")
        print(f"总行数: {len(result_df)}")
        for col in required_columns:
            non_null_count = result_df[col].notna().sum()
            print(f"  {col}: {non_null_count}/{len(result_df)} 非空")

        # 显示前几行数据示例
        print("\n前3行数据示例:")
        print(result_df.head(3))

        # 保存结果
        if save_parsed_results(result_df, output_dir, required_columns):
            print("\n处理完成!")
        else:
            print("\n保存失败!")
    else:
        print("\n处理失败或未找到数据!")


if __name__ == "__main__":
    print("脚本开始执行...")
    main()
    print("脚本执行完毕")
